/*
 * Copyright (c) 2019, Beijing Yonyou Audit Software. All rights reserved.
 *
 */
package publish;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * @author 若雨成霜
 * @date 2019-11-25
 * @description
 */
public class SubscriberPublish<M> {

    //订阅器名称
    private String name;
    //
    final int QUEUE_CAPACITY = 20;

    //订阅器存储队列
    private BlockingQueue<Msg> deque = new ArrayBlockingQueue<>(QUEUE_CAPACITY);
    //订阅者
    private List<ISubscriberPublish> subscriberPublishes = new ArrayList<>();

    public SubscriberPublish(String name) {
        this.name = name;
    }

    public void publish(String publisher, M message, boolean isInstantMsg) {
        if (isInstantMsg) {
            update(publisher, message);
            return;
        } else {
            Msg<M> m = new Msg<>(publisher, message);
            if (!deque.offer(m)) {
                update();
            }
        }
    }

    public void unSubscribe(ISubscriberPublish subscriberPublish) {
        subscriberPublishes.remove(subscriberPublish);
    }

    public void subscribe(ISubscriberPublish subscriberPublish) {
        subscriberPublishes.add(subscriberPublish);
    }

    public void update() {
        Msg msg = null;
        while ((msg = deque.peek()) != null) {
            this.update(msg.getPublisher(), (M) msg.getMsg());
        }
    }

    public void update(String publisher, M msg) {
        for (ISubscriberPublish subscriberPublish : subscriberPublishes) {
            subscriberPublish.update(publisher, msg);
        }
    }

}
